package executor

import (
	"context"
	"fmt"
	"sort"
	"strings"
	"sync"
	"time"

	"kwdb-import/pkg/parser"

	"github.com/jackc/pgx/v5"
	"github.com/jackc/pgx/v5/pgxpool"
)

// PgxStatementExecutor 基于pgx驱动的SQL语句执行器
// 这是推荐的执行器实现，符合KaiwuDB官方最佳实践
type PgxStatementExecutor struct {
	pool *pgxpool.Pool
	ctx  context.Context
}

// NewPgxStatementExecutor 创建新的pgx执行器
// connectionString 格式: postgresql://username:password@host:port/database
func NewPgxStatementExecutor(connectionString string) (*PgxStatementExecutor, error) {
	// 解析连接配置
	config, err := pgxpool.ParseConfig(connectionString)
	if err != nil {
		return nil, fmt.Errorf("解析连接配置失败: %v", err)
	}

	// 配置连接池参数
	config.MaxConns = 10                      // 最大连接数
	config.MinConns = 2                       // 最小连接数
	config.MaxConnLifetime = time.Hour        // 连接最大生存时间
	config.MaxConnIdleTime = time.Minute * 30 // 连接最大空闲时间
	config.HealthCheckPeriod = time.Minute    // 健康检查周期

	// 创建连接池
	ctx := context.Background()
	pool, err := pgxpool.NewWithConfig(ctx, config)
	if err != nil {
		return nil, fmt.Errorf("创建连接池失败: %v", err)
	}

	// 测试连接
	ctx, cancel := context.WithTimeout(ctx, 10*time.Second)
	defer cancel()

	if err := pool.Ping(ctx); err != nil {
		pool.Close()
		return nil, fmt.Errorf("数据库连接测试失败: %v", err)
	}

	return &PgxStatementExecutor{
		pool: pool,
		ctx:  context.Background(),
	}, nil
}

// Close 关闭连接池
func (e *PgxStatementExecutor) Close() error {
	if e.pool != nil {
		e.pool.Close()
	}
	return nil
}

// ExecuteStatements 执行SQL语句列表
func (e *PgxStatementExecutor) ExecuteStatements(statements []string) error {
	return e.ExecuteStatementsWithContext(e.ctx, statements)
}

// ExecuteStatementsWithContext 使用指定context执行SQL语句列表
func (e *PgxStatementExecutor) ExecuteStatementsWithContext(ctx context.Context, statements []string) error {
	// 设置超时
	ctx, cancel := context.WithTimeout(ctx, 30*time.Minute)
	defer cancel()

	// 开始事务
	tx, err := e.pool.Begin(ctx)
	if err != nil {
		return fmt.Errorf("开始事务失败: %v", err)
	}

	defer func() {
		if err != nil {
			if rollbackErr := tx.Rollback(ctx); rollbackErr != nil {
				fmt.Printf("回滚事务失败: %v\n", rollbackErr)
			}
		}
	}()

	// 执行每个语句
	for i, stmt := range statements {
		if strings.TrimSpace(stmt) == "" {
			continue
		}

		fmt.Printf("执行语句 %d/%d: %s\n", i+1, len(statements), e.truncateSQL(stmt))

		if _, err := tx.Exec(ctx, stmt); err != nil {
			return fmt.Errorf("执行语句失败 (第%d条): %v\nSQL: %s", i+1, err, stmt)
		}
	}

	// 提交事务
	if err := tx.Commit(ctx); err != nil {
		return fmt.Errorf("提交事务失败: %v", err)
	}

	return nil
}

// ExecuteStatementsWithOrder 按优先级顺序执行语句
func (e *PgxStatementExecutor) ExecuteStatementsWithOrder(statements []parser.SQLStatement, convertedStatements []string) error {
	// 创建语句和转换后语句的映射
	stmtMap := make(map[int]string)
	for i, converted := range convertedStatements {
		if i < len(statements) {
			stmtMap[i] = converted
		}
	}

	// 按优先级排序
	sortedIndices := e.sortStatementsByPriority(statements)

	// 按顺序执行
	var orderedStatements []string
	for _, index := range sortedIndices {
		if converted, exists := stmtMap[index]; exists && strings.TrimSpace(converted) != "" {
			orderedStatements = append(orderedStatements, converted)
		}
	}

	return e.ExecuteStatements(orderedStatements)
}

// ExecuteBatchInserts 批量执行INSERT语句，提高性能
func (e *PgxStatementExecutor) ExecuteBatchInserts(ctx context.Context, tableName string, columns []string, values [][]interface{}) error {
	if len(values) == 0 {
		return nil
	}

	// 构建批量INSERT语句
	placeholders := make([]string, len(columns))
	for i := range columns {
		placeholders[i] = fmt.Sprintf("$%d", i+1)
	}

	sql := fmt.Sprintf("INSERT INTO %s (%s) VALUES (%s)",
		tableName,
		strings.Join(columns, ", "),
		strings.Join(placeholders, ", "))

	// 使用批量处理
	batch := &pgx.Batch{}
	for _, row := range values {
		batch.Queue(sql, row...)
	}

	// 设置超时
	ctx, cancel := context.WithTimeout(ctx, 10*time.Minute)
	defer cancel()

	results := e.pool.SendBatch(ctx, batch)
	defer results.Close()

	// 处理结果
	for i := 0; i < len(values); i++ {
		_, err := results.Exec()
		if err != nil {
			return fmt.Errorf("批量INSERT失败 (第%d行): %v", i+1, err)
		}
	}

	return nil
}

// sortStatementsByPriority 按优先级排序语句
func (e *PgxStatementExecutor) sortStatementsByPriority(statements []parser.SQLStatement) []int {
	// 定义语句类型的执行优先级
	priority := map[parser.StatementType]int{
		parser.UseDatabase:    1, // 首先切换数据库
		parser.CreateDatabase: 2, // 创建数据库
		parser.CreateTable:    3, // 创建表
		parser.CreateSequence: 4, // 创建序列
		parser.CreateIndex:    5, // 创建索引
		parser.CreateView:     6, // 创建视图
		parser.Insert:         7, // 插入数据
		parser.Other:          8, // 其他语句
	}

	// 创建索引数组
	indices := make([]int, len(statements))
	for i := range indices {
		indices[i] = i
	}

	// 按优先级排序
	sort.Slice(indices, func(i, j int) bool {
		priorityI := priority[statements[indices[i]].Type]
		priorityJ := priority[statements[indices[j]].Type]

		if priorityI != priorityJ {
			return priorityI < priorityJ
		}

		// 相同优先级按原始顺序
		return indices[i] < indices[j]
	})

	return indices
}

// truncateSQL 截断SQL语句用于显示
func (e *PgxStatementExecutor) truncateSQL(sql string) string {
	if len(sql) > 100 {
		return sql[:97] + "..."
	}
	return sql
}

// ValidateConnection 验证数据库连接
func (e *PgxStatementExecutor) ValidateConnection() error {
	ctx, cancel := context.WithTimeout(e.ctx, 5*time.Second)
	defer cancel()
	return e.pool.Ping(ctx)
}

// GetDatabaseInfo 获取数据库信息
func (e *PgxStatementExecutor) GetDatabaseInfo() (map[string]interface{}, error) {
	ctx, cancel := context.WithTimeout(e.ctx, 10*time.Second)
	defer cancel()

	info := make(map[string]interface{})

	// 获取数据库版本
	var version string
	err := e.pool.QueryRow(ctx, "SELECT version()").Scan(&version)
	if err != nil {
		return nil, fmt.Errorf("获取数据库版本失败: %v", err)
	}
	info["version"] = version

	// 获取当前数据库名
	var dbname string
	err = e.pool.QueryRow(ctx, "SELECT current_database()").Scan(&dbname)
	if err != nil {
		return nil, fmt.Errorf("获取当前数据库名失败: %v", err)
	}
	info["database"] = dbname

	// 获取当前用户
	var user string
	err = e.pool.QueryRow(ctx, "SELECT current_user").Scan(&user)
	if err != nil {
		return nil, fmt.Errorf("获取当前用户失败: %v", err)
	}
	info["user"] = user

	// 获取连接池统计信息
	stats := e.pool.Stat()
	info["pool_stats"] = map[string]interface{}{
		"acquired_conns":      stats.AcquiredConns(),
		"constructing_conns":  stats.ConstructingConns(),
		"empty_acquire_count": stats.EmptyAcquireCount(),
		"idle_conns":          stats.IdleConns(),
		"max_conns":           stats.MaxConns(),
		"total_conns":         stats.TotalConns(),
	}

	return info, nil
}

// ExecuteSingleStatement 执行单条SQL语句
func (e *PgxStatementExecutor) ExecuteSingleStatement(sql string) error {
	ctx, cancel := context.WithTimeout(e.ctx, 30*time.Second)
	defer cancel()

	_, err := e.pool.Exec(ctx, sql)
	if err != nil {
		return fmt.Errorf("执行SQL语句失败: %v\nSQL: %s", err, sql)
	}
	return nil
}

// BeginTransaction 开始事务
func (e *PgxStatementExecutor) BeginTransaction() (pgx.Tx, error) {
	ctx, cancel := context.WithTimeout(e.ctx, 10*time.Second)
	defer cancel()
	return e.pool.Begin(ctx)
}

// GetConnectionString 构建KaiwuDB推荐的连接字符串格式
func BuildKWDBConnectionString(host, port, username, password, database string) string {
	// KaiwuDB官方推荐的连接字符串格式
	// postgresql://username:password@host:port/database
	return fmt.Sprintf("postgresql://%s:%s@%s:%s/%s?sslmode=disable",
		username, password, host, port, database)
}

// GetPoolStats 获取连接池统计信息
func (e *PgxStatementExecutor) GetPoolStats() *pgxpool.Stat {
	return e.pool.Stat()
}

// SetContext 设置执行上下文
func (e *PgxStatementExecutor) SetContext(ctx context.Context) {
	e.ctx = ctx
}

// ExecuteStatementsParallel 并发执行SQL语句列表（支持外键重试）
// workerCount: 并发worker数量
// batchSize: 每个批次的语句数量
// onError: 错误处理策略 ("continue" 或 "stop")
// maxErrors: 最大允许错误数
// 返回: 成功数, 错误数, 外键失败列表, 错误
func (e *PgxStatementExecutor) ExecuteStatementsParallel(statements []string, workerCount int, batchSize int, onError string, maxErrors int) (executedCount int, errorCount int, fkFailures []string, err error) {
	if len(statements) == 0 {
		return 0, 0, nil, nil
	}

	// 将语句分批
	batches := make([][]string, 0)
	for i := 0; i < len(statements); i += batchSize {
		end := i + batchSize
		if end > len(statements) {
			end = len(statements)
		}
		batches = append(batches, statements[i:end])
	}

	// 创建工作通道
	type batchJob struct {
		index int
		stmts []string
	}
	jobChan := make(chan batchJob, len(batches))

	// 创建结果收集通道
	type batchResult struct {
		index      int
		success    int
		errors     int
		fkErrors   []string // 外键约束失败的语句
		firstError error
	}
	resultChan := make(chan batchResult, len(batches))

	// 启动worker池
	for w := 0; w < workerCount; w++ {
		go func(workerID int) {
			for job := range jobChan {
				successCount := 0
				errCount := 0
				fkErrStmts := make([]string, 0)
				var firstErr error

				for _, stmt := range job.stmts {
					trimmed := strings.TrimSpace(stmt)
					if trimmed == "" {
						continue
					}

					if err := e.ExecuteSingleStatement(trimmed); err != nil {
						// 检查是否为外键约束错误
						if isForeignKeyError(err) {
							fkErrStmts = append(fkErrStmts, trimmed)
						} else {
							if firstErr == nil {
								firstErr = err
							}
							errCount++
						}
					} else {
						successCount++
					}
				}

				resultChan <- batchResult{
					index:      job.index,
					success:    successCount,
					errors:     errCount,
					fkErrors:   fkErrStmts,
					firstError: firstErr,
				}
			}
		}(w)
	}

	// 发送所有任务
	for i, batch := range batches {
		jobChan <- batchJob{index: i, stmts: batch}
	}
	close(jobChan)

	// 收集结果
	totalSuccess := 0
	totalErrors := 0
	allFkFailures := make([]string, 0)
	var collectedError error

	for i := 0; i < len(batches); i++ {
		result := <-resultChan
		totalSuccess += result.success
		totalErrors += result.errors
		allFkFailures = append(allFkFailures, result.fkErrors...)

		if result.firstError != nil && collectedError == nil {
			collectedError = result.firstError
		}

		// 检查是否超过最大错误数
		if totalErrors >= maxErrors {
			if onError == "stop" {
				return totalSuccess, totalErrors, allFkFailures, fmt.Errorf("错误数量超过限制 (%d): %v", maxErrors, collectedError)
			}
		}
	}

	close(resultChan)

	if totalErrors > 0 && onError == "stop" && collectedError != nil {
		return totalSuccess, totalErrors, allFkFailures, collectedError
	}

	return totalSuccess, totalErrors, allFkFailures, nil
}

// ExecuteStatementsParallelStream 并发流式执行：从通道消费语句，降低内存占用
// workerCount: 并发worker数量
// onError: 错误处理策略 ("continue" 或 "stop")
// maxErrors: 最大允许错误数
// 返回: 成功数, 错误数, 外键失败列表, 错误
func (e *PgxStatementExecutor) ExecuteStatementsParallelStream(dmlChan <-chan string, workerCount int, onError string, maxErrors int) (executedCount int, errorCount int, fkFailures []string, err error) {
	if workerCount <= 0 {
		workerCount = 1
	}

	type workerResult struct {
		success    int
		errors     int
		fkErrors   []string
		firstError error
	}

	resultChan := make(chan workerResult, workerCount)
	var wg sync.WaitGroup

	// 启动worker池，直接从dmlChan消费并执行
	for w := 0; w < workerCount; w++ {
		wg.Add(1)
		go func() {
			defer wg.Done()
			successCount := 0
			errCount := 0
			fkErrStmts := make([]string, 0)
			var firstErr error

			for stmt := range dmlChan {
				trimmed := strings.TrimSpace(stmt)
				if trimmed == "" {
					continue
				}

				if err := e.ExecuteSingleStatement(trimmed); err != nil {
					if isForeignKeyError(err) {
						fkErrStmts = append(fkErrStmts, trimmed)
					} else {
						if firstErr == nil {
							firstErr = err
						}
						errCount++
					}
				} else {
					successCount++
				}
			}

			resultChan <- workerResult{
				success:    successCount,
				errors:     errCount,
				fkErrors:   fkErrStmts,
				firstError: firstErr,
			}
		}()
	}

	// 等待所有worker结束后关闭结果通道
	go func() {
		wg.Wait()
		close(resultChan)
	}()

	totalSuccess := 0
	totalErrors := 0
	allFkFailures := make([]string, 0)
	var collectedError error

	for res := range resultChan {
		totalSuccess += res.success
		totalErrors += res.errors
		allFkFailures = append(allFkFailures, res.fkErrors...)

		if res.firstError != nil && collectedError == nil {
			collectedError = res.firstError
		}

		if totalErrors >= maxErrors && onError == "stop" && collectedError != nil {
			return totalSuccess, totalErrors, allFkFailures, fmt.Errorf("错误数量超过限制 (%d): %v", maxErrors, collectedError)
		}
	}

	if totalErrors > 0 && onError == "stop" && collectedError != nil {
		return totalSuccess, totalErrors, allFkFailures, collectedError
	}

	return totalSuccess, totalErrors, allFkFailures, nil
}
